package cn.linweijian.kafka.comsumerdemo.service;

import cn.linweijian.kafka.comsumerdemo.common.Result;
import cn.linweijian.kafka.comsumerdemo.vo.Article;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * Description: 发送消息
 * <p>
 * Created by linweijian on 2018/5/31 11:47
 */
@Component
public class ReceiverService {
    private final Logger log = LoggerFactory.getLogger(ReceiverService.class);
    private Gson gson = new GsonBuilder().create();

    /**
     * 接收消息
     *
     * @param record
     * @return void
     * @author linweijian
     * @date 2018/5/31 11:48
     **/
    @KafkaListener(topics = {"zhuti1"})
    public void listen(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            String message = (String)kafkaMessage.get();
            log.info("----------------- record =" + record);
            log.info("----------------- key =" + record.key());
            log.info("------------------ message =" + message);
           Article article = gson.fromJson(message,Article.class);
           System.out.println(article);
           Result.getList().add(article);
        }

    }
}
